package com.sparrow.rpc.socket.handler;

import com.sparrow.rpc.manager.RpcNodeManager;
import com.sparrow.rpc.socket.msg.req.RpcHeatBeat;
import com.sparrow.rpc.socket.msg.req.RpcLoginRequest;
import com.sparrow.rpc.socket.msg.req.RpcRequest;
import com.sparrow.rpc.service.RpcService;
import com.sparrow.rpc.socket.msg.resp.RpcLoginResponse;
import com.sparrow.rpc.socket.session.RpcServerSessionManager;
import com.sparrow.rpc.socket.session.RpcSession;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.Attribute;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

@ChannelHandler.Sharable
public class RpcServerHandler extends ChannelInboundHandlerAdapter {
    private static Logger logger = LoggerFactory.getLogger(RpcServerHandler.class);

    @Autowired
    private RpcService rpcService;
    @Autowired
    private RpcServerSessionManager sessionManager;
    @Autowired
    private RpcNodeManager nodeManager;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        RpcSession session = new RpcSession();
        session.setChannel(ctx.channel());
        ctx.channel().attr(RpcSession.SESSION).set(session);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        Attribute<RpcSession> attr = ctx.channel().attr(RpcSession.SESSION);
        RpcSession rpcSession = attr.get();
        if (rpcSession != null) {
            sessionManager.close(rpcSession);
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        RpcSession session = ctx.channel().attr(RpcSession.SESSION).get();
        if (msg instanceof RpcLoginRequest) {
            RpcLoginRequest loginRequest = (RpcLoginRequest) msg;
            sessionManager.bind(session, loginRequest.getNodeId());

            RpcLoginResponse loginResp = new RpcLoginResponse();
            loginResp.setTargetNodeId(nodeManager.getSelf().getId());
            loginResp.setLoginNodeId(loginRequest.getNodeId());
            session.writeAndFlush(loginResp);

            nodeManager.registerNode(loginRequest.getNodeId(), loginRequest.getIp(), loginRequest.getPort());

            nodeManager.onRpcNodeLogin(session, loginRequest);
        } else if (msg instanceof RpcRequest) {
            RpcRequest rpcRequest = (RpcRequest) msg;
            rpcService.request(session, rpcRequest);
        } else if (msg instanceof RpcHeatBeat) {
            //心跳不用处理
            if (logger.isDebugEnabled()) {
                logger.debug("rpc server 收到心跳 channel:{}", ctx.channel().toString());
            }
        }
    }

}
